//package com.wdd.myplatform.config.pulsar;
//
//import org.apache.pulsar.client.api.Producer;
//import org.apache.pulsar.client.api.ProducerBuilder;
//import org.apache.pulsar.client.api.PulsarClient;
//import org.apache.pulsar.client.api.PulsarClientException;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//
//import java.util.concurrent.TimeUnit;
//
///**
// * @author wdd 2022/7/18
// */
//@Configuration
//public class PulsarProducerConfig {
//    @Value("${producer.name:producername}")
//    String producerName;
//
//    @Value("${topic.name:airquality}")
//    String topicName;
//
//    @Autowired
//    PulsarClient pulsarClient;
//
//    /*
//    * Producer 的参数:
//    1.topic:Producer 要写入的 topic。
//    2.compressionType:压缩策略,目前支持 4 种策略 (NONE、LZ4、ZLIB、ZSTD)
//    3.sendTimeout：超时时间。如果 Producer 在超时时间为收到 ACK,会重复发送
//    4.enableBatching：是否开启消息批量处理。这里默认 true，这个参数只有在异步发送 (sendAsync) 时才能生效，选择同步发送会失效。
//    5.batchingMaxPublishDelay：批量发送消息的时间段，这里定义的是 10ms，需要注意的是，设置了批量时间，就不会受消息数量的影响。批量发送会把要发送的批量消息放在一个网络包里发送出去，减少网络 IO 次数，大大提高网卡的发送效率。
//    6.batchingMaxMessages：批量发送消息的最大数量。
//    7.maxPendingMessages：等待从 broker 接收 ACK 的消息队列最大长度。如果这个队列满了，producer 所有的 sendAsync 和 send 都会失败，除非设置了 blockIfQueueFull 值是 true。
//    8.blockIfQueueFull：Producer 发送消息时会把消息先放入本地 Queue 缓存，如果缓存满了，就会阻塞消息发送。
//    9.roundRobinRouterBatchingPartition-SwitchFrequency：如果发送消息时没有指定 key，那默认采用 round robin 的方式发送消息，使用 round robin 的方式，切换 partition 的周期是 (frequency * batchingMaxPublishDelay)。
//    * */
//    @Bean
//    public Producer getProducer() {
//        ProducerBuilder producerBuilder = pulsarClient.newProducer()
//                .topic(topicName)
//                .producerName(producerName).sendTimeout(60, TimeUnit.SECONDS);
//
//        Producer producer = null;
//        try {
//            producer = producerBuilder.create();
//        } catch (PulsarClientException e1) {
//            e1.printStackTrace();
//        }
//        return producer;
//    }
//}
